// Copyright 2013 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.google.api.ads.dfa.matchtables;

import com.google.api.ads.dfa.axis.factory.DfaServices;
import com.google.api.ads.dfa.axis.v1_20.ApiException;
import com.google.api.ads.dfa.axis.v1_20.CreativeBase;
import com.google.api.ads.dfa.axis.v1_20.CreativeField;
import com.google.api.ads.dfa.axis.v1_20.CreativeFieldAssignment;
import com.google.api.ads.dfa.axis.v1_20.CreativeFieldRecordSet;
import com.google.api.ads.dfa.axis.v1_20.CreativeFieldRemote;
import com.google.api.ads.dfa.axis.v1_20.CreativeFieldSearchCriteria;
import com.google.api.ads.dfa.axis.v1_20.CreativeFieldValue;
import com.google.api.ads.dfa.axis.v1_20.CreativeFieldValueRecordSet;
import com.google.api.ads.dfa.axis.v1_20.CreativeFieldValueSearchCriteria;
import com.google.api.ads.dfa.axis.v1_20.CreativeRecordSet;
import com.google.api.ads.dfa.axis.v1_20.CreativeRemote;
import com.google.api.ads.dfa.axis.v1_20.CreativeSearchCriteria;
import com.google.api.ads.dfa.axis.v1_20.ImageCreativeBase;
import com.google.api.ads.dfa.axis.v1_20.RedirectCreativeBase;
import com.google.api.ads.dfa.axis.v1_20.RichMediaCreativeBase;
import com.google.api.ads.dfa.axis.v1_20.RichMediaEventBase;
import com.google.api.ads.dfa.axis.v1_20.RichMediaStandardEvent;
import com.google.api.ads.dfa.lib.client.DfaSession;

import com.google.api.ads.dfa.matchtables.CreativeTypes.SoapCreativeType;

import java.rmi.RemoteException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/**
 * Implementation of LookupObject that will generate Creative match tables. Designed to be
 * component of AdCreativeLookup to be able to exchange data with AdSubLookup.
 *
 * @author Moshe Kahan
 */
public class CreativeSubLookup implements LookupObject {

  // Main aggregation maps
  private Map<Long, Set<Long>> advertiserToRenderingIdsMap;
  private Set<Long> metEventIdSet;

  // DT columns that this class is interested in
  private int advertiserColumnIndex;
  private int creativeColumnIndex;
  private int metEventColumnIndex;

  // MatchTable settings and SOAP remote objects
  private TableSettings tableSettings;
  private final ThreadLocal<CreativeRemote> creativeService;
  private final ThreadLocal<CreativeFieldRemote> creativeFieldService;

  // Secondary lookup maps
  private Map<Long, Long> creativeToRenderingMap;
  // Size helper
  SizeLookupHelper sizeLookupHelper;

  // Multi-Threading support.
  private ExecutorService executor;

  /**
   * Constructor.  Besides the DFA service/session objects needed to create SOAP remote objects to
   * the API, constructor also accepts matchTableTypes to determine if both Ad and Creative match
   * tables should be generated.
   *
   * @param dfaServices DFA services object needed to create SOAP remotes to the API
   * @param session DFA session object needed to create SOAP remotes to the API
   * @param tableSettings match table application configuration used as a MatchTable object factory
   */
  CreativeSubLookup(final DfaServices dfaServices, final DfaSession session,
      TableSettings tableSettings, ExecutorService executor, SizeLookupHelper sizeLookupHelper,
      Map<Long, Long> creativeToRenderingMap) {
    this(AdCreativeLookup.initializeService(dfaServices, session, CreativeRemote.class),
        AdCreativeLookup.initializeService(dfaServices, session, CreativeFieldRemote.class),
        tableSettings, executor, sizeLookupHelper, creativeToRenderingMap);
  }

  CreativeSubLookup(ThreadLocal<CreativeRemote> creativeService,
      ThreadLocal<CreativeFieldRemote> creativeFieldService, TableSettings tableSettings,
      ExecutorService executor, SizeLookupHelper sizeLookupHelper,
      Map<Long, Long> creativeToRenderingMap) {
    // Request the advertiser service from the service client factory.
    advertiserToRenderingIdsMap = new HashMap<Long, Set<Long>>();
    metEventIdSet = new HashSet<Long>();

    this.creativeService = creativeService;
    this.creativeFieldService = creativeFieldService;
    this.creativeToRenderingMap = creativeToRenderingMap;
    this.tableSettings = tableSettings;
    // Set up Executor to run concurrent DFA API retrievals.
    this.executor = executor;
    // Size lookup helper
    this.sizeLookupHelper = sizeLookupHelper;

    advertiserColumnIndex = -1;
    creativeColumnIndex = -1;
    metEventColumnIndex = -1;
  }

  /**
   * Called when parsing the DT file header. Method stores index of creative column.
   *
   * @param index in DT file columns
   * @param header name of this DT file field/column
   */
  public void addHeader(int index, String header) {
    if (LookupField.AdvertiserId.getFieldName().equals(header)) {
      advertiserColumnIndex = index;
    } else if (LookupField.CreativeId.getFieldName().equals(header) && creativeColumnIndex < 0) {
      // Only use CreativeID if RenderingID is not present
      // since RM DT's RenderingID == Non-RM DT's CreativeID.
      creativeColumnIndex = index;
    } else if (LookupField.RenderingId.getFieldName().equals(header)) {
      creativeColumnIndex = index;
    } else if (LookupField.MetEventId.getFieldName().equals(header)) {
      metEventColumnIndex = index;
    }
  }

  /**
   * Called to clear the DT header information' usually because you're processing a new file.
   */
  public void clearHeaders() {
    advertiserColumnIndex = -1;
    creativeColumnIndex = -1;
    metEventColumnIndex = -1;
  }

  /**
   * Called after DT file line has been tokenized into array of Strings.
   *
   * @param values array of tokenized Strings of DT file line
   */
  public void addValues(String[] values) {
    // Can only parse if Ad-ID, Advertiser-ID, and Buy-ID are present.
    if (advertiserColumnIndex > -1) {
      String advertiserId = values[advertiserColumnIndex];
      if (creativeColumnIndex > -1) {
        String renderingId = values[creativeColumnIndex];
        if (AdCreativeLookup.isValid(advertiserId) && AdCreativeLookup.isValid(renderingId)) {
          AdCreativeLookup.addToSetToMap(
              advertiserToRenderingIdsMap, Long.valueOf(advertiserId), Long.valueOf(renderingId));
          if (metEventColumnIndex > -1) {
            String metEventId = values[metEventColumnIndex];
            if (AdCreativeLookup.isValid(metEventId)) {
              metEventIdSet.add(Long.valueOf(metEventId));
            }
          }
        }
      }
    }
  }

  /**
   * Called when parsing/aggregation is complete. Will lookup via SOAP all unique DFA entries
   * in the DT file and pass data to MatchTable class.
   *
   * @param outputPath location of match table to be generated.
   */
  public void generateMatchTable(String outputPath)
      throws MatchTableFileException, DfaRemoteException {
    List<MatchTable> matchTableList = new ArrayList<MatchTable>();
    MatchTable creativeMatchTable =
        AdCreativeLookup.createMatchTable(tableSettings, MatchTable.Type.CREATIVE, matchTableList);
    MatchTable creativeCustomFieldMatchTable = AdCreativeLookup.createMatchTable(tableSettings,
        MatchTable.Type.CUSTOM_CREATIVE_FIELDS, matchTableList);
    MatchTable metEventMatchTable = AdCreativeLookup.createMatchTable(tableSettings,
        MatchTable.Type.MET_EVENT, matchTableList);
    for (MatchTable matchTable : matchTableList) {
      matchTable.open(outputPath);
    }

    // If we have more than 1 than we know it's not just campaignMatchTable; or if
    // the 1 table we have is not just campaignMatchTable
    Progress.printProgress(creativeMatchTable, creativeCustomFieldMatchTable,
        metEventMatchTable);

    System.out.println("Creative Retrieval Phase");
    // Retrieve raw creative entities concurrently.
    List<Future<?>> taskResults = new ArrayList<Future<?>>();
    final Progress progress = new Progress();
    progress.start(advertiserToRenderingIdsMap.keySet().size());
    final List<CreativeBase> creativeResults = new ArrayList<CreativeBase>();
    // Rich Media Results (will be populated by generateCreativeMatchTable)
    Map<Long, Set<RichMediaEventBase>>  advertiserRichMediaSetMap =
        new HashMap<Long, Set<RichMediaEventBase>>();
    for (Map.Entry<Long, Set<Long>> entry : advertiserToRenderingIdsMap.entrySet()) {
      final long advertiserId = entry.getKey();
      final Set<Long> creativeIdSet = entry.getValue();
      Set<RichMediaEventBase> metEvents = new HashSet<RichMediaEventBase>();
      advertiserRichMediaSetMap.put(advertiserId, metEvents);
      taskResults.add(executor.submit(new Callable<Void>() {
        public Void call() throws DfaRemoteException {
          retrieveCreatives(advertiserId, creativeIdSet, creativeResults);
          progress.next();
          return null;
        }
      }));
    }
    AdCreativeLookup.waitForFutures(taskResults);
    progress.end();

    // Size lookup
    sizeLookupHelper.retrieveCreativeSizes(creativeResults);

    // Generate creative match tables. Store RichMedia entities into advertiserRichMediaSetMap.
    // Now process all creatives in this campaign for these ads.
    generateCreativeMatchTable(creativeResults, advertiserRichMediaSetMap, creativeMatchTable,
        creativeCustomFieldMatchTable);

    System.out.println("Creative Secondary Retrieval Phase");
    if (metEventMatchTable != null) {
      generateMetEventTable(advertiserRichMediaSetMap, metEventMatchTable);
    }

    for (MatchTable matchTable : matchTableList) {
      matchTable.close();
    }
  }

  // Generate all creative match tables for this advertiser and campaign.
  private void generateCreativeMatchTable(List<CreativeBase> creatives,
    Map<Long, Set<RichMediaEventBase>> metAdvertiserEvents,
    MatchTable creativeMatchTable,
    MatchTable creativeCustomFieldMatchTable)
        throws MatchTableFileException, DfaRemoteException {
    Map<Long, CreativeField> creativeFields = new HashMap<Long, CreativeField>();
    Map<Long, String> creativeFieldValues = new HashMap<Long, String>();
    Map<Long, CreativeFieldAssignment[]> creativeFieldAssignmentMap =
        new HashMap<Long, CreativeFieldAssignment[]>();

    SimpleDateFormat dateFormat = new SimpleDateFormat("MM-dd-yyyy");
    for (CreativeBase result : creatives) {
      creativeToRenderingMap.put(result.getId(), result.getRenderingId());
      CreativeFieldAssignment[] creativeFieldAssignments = result.getCreativeFieldAssignments();
      if (creativeFieldAssignments != null && creativeFieldAssignments.length > 0) {
        creativeFieldAssignmentMap.put(result.getId(), creativeFieldAssignments);
        for (CreativeFieldAssignment creativeFieldAssignment : creativeFieldAssignments) {
          creativeFields.put(creativeFieldAssignment.getCreativeFieldId(), null);
          creativeFieldValues.put(creativeFieldAssignment.getCreativeFieldValueId(), null);
        }
      }
      extractMetEvents(result, metAdvertiserEvents.get(result.getAdvertiserId()));

      SoapCreativeType soapCreativeType = SoapCreativeType.getById(result.getTypeId());
      String renderingId = Long.toString(result.getRenderingId());

      // store custom field IDs
      if (creativeMatchTable != null) {
        String modifiedDate = "";
        if (result.getModifiedDate() != null) {
          modifiedDate = dateFormat.format(result.getModifiedDate().getTime());
        }
        creativeMatchTable.write(
            Long.toString(result.getAdvertiserId()),
            renderingId,
            Long.toString(result.getId()),
            result.getName(),
            modifiedDate,
            soapCreativeType != null ?
                soapCreativeType.getMatchTableCreativeType().getName() : "",
            soapCreativeType != null ?
                soapCreativeType.getMatchTableCreativeSubType().getName() : "",
            sizeLookupHelper.getSize(result.getSizeId()),
            extractImageURL(result),
            Long.toString(result.getVersion()));
      }
    }
    if (creativeMatchTable != null) {
      creativeMatchTable.flush();
    }
    generateCreativeCustomFieldMatchTable(creativeFields, creativeFieldValues,
        creativeFieldAssignmentMap, creativeCustomFieldMatchTable);
  }

  private void generateCreativeCustomFieldMatchTable(Map<Long, CreativeField> creativeFields,
      Map<Long, String> creativeFieldValues,
      Map<Long, CreativeFieldAssignment[]> creativeFieldAssignmentMap,
      MatchTable creativeCustomFieldMatchTable)
          throws MatchTableFileException, DfaRemoteException {
    // Look up custom fields and generate match table for them.
    if (creativeCustomFieldMatchTable != null) {
      SoapPageableLookup<CreativeField, CreativeFieldSearchCriteria> creativeFieldSoapLookup =
          new SoapPageableLookup<CreativeField, CreativeFieldSearchCriteria>(
              creativeFields.keySet(), new CreativeFieldSoapLookupCallback());
      while (creativeFieldSoapLookup.hasNext()) {
        CreativeField result = creativeFieldSoapLookup.next();
        creativeFields.put(result.getId(), result);
      }

      SoapPageableLookup<CreativeFieldValue, CreativeFieldValueSearchCriteria>
      creativeFieldValueSoapLookup =
        new SoapPageableLookup<CreativeFieldValue, CreativeFieldValueSearchCriteria>(
            creativeFieldValues.keySet(), new CreativeFieldValueSoapLookupCallback());
      while (creativeFieldValueSoapLookup.hasNext()) {
        CreativeFieldValue result = creativeFieldValueSoapLookup.next();
        creativeFieldValues.put(result.getId(), result.getName());
      }

      for (Map.Entry<Long, CreativeFieldAssignment[]> entry :
        creativeFieldAssignmentMap.entrySet()) {
        for (CreativeFieldAssignment assignment : entry.getValue()) {
          String renderingId =
              Long.toString(creativeToRenderingMap.get(entry.getKey()));
          CreativeField creativeField =
              creativeFields.get(assignment.getCreativeFieldId());
          String creativeFieldValue =
              creativeFieldValues.get(assignment.getCreativeFieldValueId());
          creativeCustomFieldMatchTable.write(
              Long.toString(creativeField.getAdvertiserId()),
              renderingId,
              Long.toString(creativeField.getSlotNumber()),
              creativeField.getName(),
              creativeFieldValue);
        }
      }
    }
    if (creativeCustomFieldMatchTable != null) {
      creativeCustomFieldMatchTable.flush();
    }
  }

  private void generateMetEventTable(Map<Long, Set<RichMediaEventBase>> advertiserRichMediaSetMap,
      MatchTable metEventMatchTable) throws MatchTableFileException, DfaRemoteException {

    Map<String, String> richMediaEventTypes = new HashMap<String, String>();
    richMediaEventTypes.put("Timer", "1");
    richMediaEventTypes.put("Exit", "2");
    richMediaEventTypes.put("Counter", "3");

    // Retrieve standard MET events.
    List<RichMediaStandardEvent> standardEvents = new ArrayList<RichMediaStandardEvent>();
    retriveMetStandardEvents(standardEvents);

    for (RichMediaStandardEvent standardEvent : standardEvents) {
      if (metEventIdSet.contains(standardEvent.getMetEventId())) {
        metEventMatchTable.write(
            "0",
            Long.toString(standardEvent.getMetEventId()),
            standardEvent.getName(),
            Long.toString(standardEvent.getId()),
            standardEvent.getName());
      }
    }
    metEventMatchTable.flush();

    // Generate custom MET events now.
    for (Map.Entry<Long, Set<RichMediaEventBase>> entry : advertiserRichMediaSetMap.entrySet()) {
      for (RichMediaEventBase richMediaEventBase : entry.getValue()) {
        metEventMatchTable.write(
            Long.toString(entry.getKey()),
            Long.toString(richMediaEventBase.getMetEventId()),
            richMediaEventBase.getName(),
            richMediaEventTypes.get(richMediaEventBase.getType()),
            richMediaEventBase.getType());
      }
    }
    metEventMatchTable.flush();
  }

  //Thread-Safe
  private void retrieveCreatives(long advertiserId, Set<Long> creativeIds,
      List<CreativeBase> creativeResults) throws DfaRemoteException {
    List<CreativeBase> creatives = new ArrayList<CreativeBase>();
    SoapPageableLookup<CreativeBase, CreativeSearchCriteria> soapLookup =
        new SoapPageableLookup<CreativeBase, CreativeSearchCriteria>(creativeIds,
            new CreativeSoapLookupCallback(advertiserId));
    while (soapLookup.hasNext()) {
      CreativeBase result  = soapLookup.next();
      if (result != null) {
        creatives.add(result);
      }
    }
    addCreatives(creativeResults, creatives);
  }

  private void retriveMetStandardEvents(List<RichMediaStandardEvent> standardEvents)
      throws DfaRemoteException {
    SoapLookup<RichMediaStandardEvent> soapLookup = new SoapLookup<RichMediaStandardEvent>(
        new RichMediaStandardEventLookupCallback());
    while (soapLookup.hasNext()) {
      RichMediaStandardEvent result = soapLookup.next();
      if (result != null) {
        standardEvents.add(result);
      }
    }
  }

  private void addCreatives(List<CreativeBase> creativeResults, List<CreativeBase> creatives) {
    synchronized (creativeResults) {
      creativeResults.addAll(creatives);
    }
  }

  // Extract image URL property if downcasted Creative type supports that property.
  private String extractImageURL(CreativeBase creative) {
    if (creative instanceof ImageCreativeBase) {
      return ((ImageCreativeBase) creative).getAssetFilename();
    } else if (creative instanceof RedirectCreativeBase) {
      return ((RedirectCreativeBase) creative).getRedirectUrl();
    }
    return null;
  }

  // Extract RM events from creative if downcasted Creative type supports that property.
  private void extractMetEvents(CreativeBase creative, Set<RichMediaEventBase> metEvents) {
    if (creative instanceof RichMediaCreativeBase) {
      RichMediaCreativeBase richMediaCreativeBase = (RichMediaCreativeBase) creative;
      for (RichMediaEventBase richMediaEvent : richMediaCreativeBase.getCounterEvents()) {
        addMetEvent(metEvents, richMediaEvent);
      }
      for (RichMediaEventBase richMediaEvent : richMediaCreativeBase.getExitEvents()) {
        addMetEvent(metEvents, richMediaEvent);
      }
      for (RichMediaEventBase richMediaEvent : richMediaCreativeBase.getTimerEvent()) {
        addMetEvent(metEvents, richMediaEvent);
      }
    }
  }

  private void addMetEvent(Set<RichMediaEventBase> metEvents, RichMediaEventBase richMediaEvent) {
    if (metEventIdSet.contains(richMediaEvent.getMetEventId())) {
      metEvents.add(richMediaEvent);
    }
  }

  private class CreativeSoapLookupCallback
    implements SoapPageableLookupCallback<CreativeBase, CreativeSearchCriteria> {

    CreativeSoapLookupCallback(long advertiserId) {
      this.advertiserId = advertiserId;
    }

    public int getRecords(CreativeSearchCriteria criteria, List<CreativeBase> recordList)
        throws ApiException, RemoteException {

      CreativeRecordSet creativeRecordSet = creativeService.get().getCreatives(criteria);
      for (CreativeBase entity : creativeRecordSet.getRecords()) {
        recordList.add(entity);
      }
      return creativeRecordSet.getTotalNumberOfPages();
    }

    public CreativeSearchCriteria getCriteria(long[] ids) {
      CreativeSearchCriteria creativeSearchCriteria = new CreativeSearchCriteria();
      creativeSearchCriteria.setAdvertiserId(advertiserId);
      creativeSearchCriteria.setRenderingIds(ids);
      return creativeSearchCriteria;
    }

    private long advertiserId;
  }

  private class CreativeFieldSoapLookupCallback
    implements SoapPageableLookupCallback<CreativeField, CreativeFieldSearchCriteria> {

    CreativeFieldSoapLookupCallback() {}

    public int getRecords(CreativeFieldSearchCriteria criteria, List<CreativeField> recordList)
        throws ApiException, RemoteException {
      CreativeFieldRecordSet creativeFieldRecordSet =
          creativeFieldService.get().getCreativeFields(criteria);
      for (CreativeField entity : creativeFieldRecordSet.getRecords()) {
        recordList.add(entity);
      }
      return creativeFieldRecordSet.getTotalNumberOfPages();
    }

    public CreativeFieldSearchCriteria getCriteria(long[] ids) {
      CreativeFieldSearchCriteria searchCriteria = new CreativeFieldSearchCriteria();
      if (ids != null) {
        searchCriteria.setIds(ids);
      }
      return searchCriteria;
    }
  }

  private class CreativeFieldValueSoapLookupCallback
    implements SoapPageableLookupCallback<CreativeFieldValue, CreativeFieldValueSearchCriteria> {

    CreativeFieldValueSoapLookupCallback() {}

    public int getRecords(CreativeFieldValueSearchCriteria criteria,
        List<CreativeFieldValue> recordList)
        throws ApiException, RemoteException {
      CreativeFieldValueRecordSet recordSet =
          creativeFieldService.get().getCreativeFieldValues(criteria);
      for (CreativeFieldValue entity : recordSet.getRecords()) {
        recordList.add(entity);
      }
      return recordSet.getTotalNumberOfPages();
    }

    public CreativeFieldValueSearchCriteria getCriteria(long[] ids) {
      CreativeFieldValueSearchCriteria searchCriteria = new CreativeFieldValueSearchCriteria();
      if (ids != null) {
        searchCriteria.setIds(ids);
      }
      return searchCriteria;
    }
  }

  private class RichMediaStandardEventLookupCallback
    implements SoapLookupCallback<RichMediaStandardEvent> {

    RichMediaStandardEventLookupCallback() {}

    public int getRecords(long[] ids, List<RichMediaStandardEvent> recordList)
        throws ApiException, RemoteException {
      RichMediaStandardEvent[] standardEvents = creativeService.get().getRichMediaStandardEvents();
      if (standardEvents != null) {
        for (RichMediaStandardEvent standardEvent : standardEvents) {
          recordList.add(standardEvent);
        }
        return standardEvents.length;
      }
      return 0;
    }
  }
}
